package demo.canal;

import com.google.protobuf.InvalidProtocolBufferException;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;

import java.net.InetSocketAddress;
import java.util.List;

public class CanalClient implements InitializingBean {

    Logger logger = LoggerFactory.getLogger(CanalClient.class);

    @Override
    public void afterPropertiesSet() {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.8.190", 11111), "example", "", "");
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            while (true) {
                Message message = connector.getWithoutAck(1000);
                long batchId = message.getId();
                List<CanalEntry.Entry> entries = message.getEntries();
                int size = entries.size();
                if (batchId == -1 || size == 0) {
                    logger.info("休眠。。。。");
                    Thread.sleep(2000);
                } else {
                    entries.forEach(item -> {
                        CanalEntry.EntryType entryType = item.getEntryType();
                        if (!(entryType == CanalEntry.EntryType.TRANSACTIONBEGIN || entryType == CanalEntry.EntryType.TRANSACTIONEND)) {
                            try {
                                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(item.getStoreValue());
                                CanalEntry.EventType eventType = rowChange.getEventType();
                                logger.info("事件类型：" + eventType);
                                logger.info("数据信息：binlog[%s:%s],name[%s,%s],envent[%s]",
                                            item.getHeader().getLogfileName(),item.getHeader().getLogfileOffset()
                                           ,item.getHeader().getSchemaName(),item.getHeader().getTableName(),eventType);
                                List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
                                rowDatasList.forEach(data -> {
                                    if (eventType == CanalEntry.EventType.DELETE) {
                                        List<CanalEntry.Column> beforeColumnsList = data.getBeforeColumnsList();
                                        System.out.println("=======================【删除事件】======================");
                                        beforeColumnsList.forEach(column -> {
                                            logger.info("字段[%s],value[%s]",column.getName(),column.getValue());
                                        });
                                        //如果是新增语句
                                    } else if (eventType == CanalEntry.EventType.INSERT) {
                                        System.out.println("=======================【插入事件】======================");
                                        List<CanalEntry.Column> afterColumnsList = data.getAfterColumnsList();
                                        afterColumnsList.forEach(column -> {
                                            logger.info("字段[%s],value[%s]",column.getName(),column.getValue());
                                        });
                                    } else {
                                        //变更前的数据
                                        System.out.println("=======================【更新事件】======================");
                                        List<CanalEntry.Column> beforeColumnsList = data.getBeforeColumnsList();
                                        System.out.println("------->; before");
                                        beforeColumnsList.forEach(column -> {
                                            logger.info("字段[%s],value[%s]",column.getName(),column.getValue());
                                        });
                                        //变更后的数据
                                        List<CanalEntry.Column> afterColumnsList = data.getAfterColumnsList();
                                        System.out.println("------->; after");
                                        afterColumnsList.forEach(column -> {
                                            logger.info("字段[%s],value[%s]",column.getName(),column.getValue());
                                        });
                                    }
                                });
                            } catch (InvalidProtocolBufferException e) {
                                e.printStackTrace();
                            }
                        }
                    });
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
